package com.wei.violet.app

import com.wei.violet.config.ConfigHandle
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}

object ChannelHandle {
  Logger.getLogger("org").setLevel(Level.WARN)
  def main(args: Array[String]): Unit = {



  val conf: SparkConf = new SparkConf()
    .setAppName("将终端设备统计结果添加到mysql数据库中")
    .setMaster("local[*]")
    .set("spark.serizlizer", "org.apache.spark.serializer.KryoSerializer")

  val sc = new SparkContext(conf)
  val sQLContext = new SQLContext(sc)
  val dataFrame: DataFrame = sQLContext.read.parquet("F:\\bigdataFile\\parquet")
  dataFrame.registerTempTable("localdistribution")
  val sqlResult = sQLContext.sql(
    """
      |select
      |    channelid
      |, SUM (CASE WHEN requestmode =1 and processnode >=1 THEN 1 ELSE 0 END) as InitialRequest,
      |SUM (CASE WHEN requestmode =1 and processnode >=2 THEN 1 ELSE 0 END) as EffectiveRequest,
      |SUM (CASE WHEN requestmode =1 and processnode =3 THEN 1 ELSE 0 END) as ADRequests,
      |SUM (CASE WHEN iseffective =1 and isbilling =1 and isbid=1 and adorderid !=0 THEN 1 ELSE 0 END) as BiddingCount,
      |SUM (CASE WHEN requestmode =2 and  iseffective =1 and isbid=1 and adorderid !=0 THEN 1 ELSE 0 END) as ADShow,
      |SUM (CASE WHEN requestmode =3 and  iseffective =1 and isbid=1 and adorderid !=0 THEN 1 ELSE 0 END) as ADHits,
      |SUM (CASE WHEN iseffective =1 and isbilling =1 and iswin=1 THEN winprice ELSE 0 END) as DSPADExpense,
      |SUM (CASE WHEN iseffective =1 and isbilling =1 and iswin=1 THEN adpayment ELSE 0 END) as DSPADCosting,
      |SUM (CASE WHEN iseffective =1 and isbilling =1 and iswin=1 THEN 1 ELSE 0 END) as BiddingSuccess
      |from localdistribution
      |group by channelid
      |
      """.stripMargin)

  sqlResult.write.mode(SaveMode.Overwrite).jdbc(

    ConfigHandle.url,
    "channelsparksql",
    ConfigHandle.conn


  )


  sc.stop()

}

}
